package com.gomro.galaxy.foundation.bi.acquisition.job;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.gomro.galaxy.core.batch.ElasticSearchClzItemWriter;
import com.gomro.galaxy.core.commons.elastic.ElasticSearchRest;
import com.gomro.galaxy.foundation.bi.domain.Constants;
import com.gomro.galaxy.foundation.bi.domain.enumerater.AcquireBatchStatus;
import com.gomro.galaxy.foundation.bi.domain.enumerater.AcquirePlan;
import com.gomro.galaxy.foundation.bi.domain.event.BiAcquireBatchEvent;
import com.gomro.galaxy.foundation.bi.domain.model.AcquireLastBatch;
import com.gomro.galaxy.foundation.bi.domain.model.AcquireOrderGoods;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.batch.core.*;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.database.JdbcPagingItemReader;
import org.springframework.batch.item.database.Order;
import org.springframework.batch.item.database.builder.JdbcPagingItemReaderBuilder;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cloud.bus.BusProperties;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.lang.NonNull;

import javax.sql.DataSource;
import java.util.*;
//订单商品
@Slf4j
@Configuration
public class OrderGoodsJob {

    private static final String BiAcquireLastName = Constants.BiWarehouseIndexPrefix + AcquireLastBatch.TableName;
    private static final String BiWarehouseName = Constants.BiWarehouseIndexPrefix + AcquireOrderGoods.TableName;
    private static final String JobName = Constants.JobBeanNamePrefix + AcquireOrderGoods.TableName;
    private static final String StepName = Constants.StepBeanNamePrefix + AcquireOrderGoods.TableName;
    private static final String ReaderName = Constants.ReaderBeanNamePrefix + AcquireOrderGoods.TableName;


    RowMapper<AcquireOrderGoods> rowMapper = (rs, i) -> AcquireOrderGoods.builder()
            .id(rs.getLong("id"))
            .vendorOrderId(rs.getLong("vendor_order_id"))
            .cartSrc(rs.getInt("cart_src"))
            .goodsId(rs.getLong("goods_id"))
            .category(rs.getString("category"))
            .brand(rs.getString("brand"))
            .name(rs.getString("name"))
            .model(rs.getString("model"))
            .spec(rs.getString("spec"))
            .sku(rs.getString("sku"))
            .barCode(rs.getString("bar_code"))
            .goldenTax(rs.getString("golden_tax"))
            .unspsc(rs.getString("unspsc"))
            .images(rs.getString("images"))
            .description(rs.getString("description"))
            .technical(rs.getString("technical"))
            .packaging(rs.getString("packaging"))
            .memo(rs.getString("memo"))
            .unit(rs.getString("unit"))
            .packageUnit(rs.getString("package_unit"))
            .packageNum(rs.getString("package_num"))
            .warranty(rs.getString("warranty"))
            .delivery(rs.getInt("delivery"))
            .length(rs.getDouble("length"))
            .width(rs.getDouble("width"))
            .height(rs.getDouble("height"))
            .weight(rs.getDouble("weight"))
            .num(rs.getDouble("num"))
            .pretaxPrice(rs.getDouble("pretax_price"))
            .price(rs.getDouble("price"))
            .pretaxFreight(rs.getDouble("pretax_freight"))
            .taxRateFreight(rs.getDouble("tax_rate_freight"))
            .taxPayableFreight(rs.getDouble("tax_payable_freight"))
            .freight(rs.getDouble("freight"))
            .taxRate(rs.getDouble("tax_rate"))
            .taxPayable(rs.getDouble("tax_payable"))
            .pretaxTotal(rs.getDouble("pretax_total"))
            .total(rs.getDouble("total"))
            .pretaxAmount(rs.getDouble("pretax_amount"))
            .amount(rs.getDouble("amount"))
            .customerAccountPeriod(rs.getInt("customer_account_period"))
            .vendorAccountPeriod(rs.getInt("vendor_account_period"))
            .customerPeriodTime(rs.getTimestamp("customer_period_time").toLocalDateTime())
            .vendorPeriodTime(rs.getTimestamp("vendor_period_time").toLocalDateTime())
            .status(rs.getInt("status"))
            .del(rs.getBoolean("del"))
            .last(rs.getTimestamp("last")==null?rs.getTimestamp("time").toLocalDateTime():rs.getTimestamp("last").toLocalDateTime())
            .time(rs.getTimestamp("time").toLocalDateTime())
            .operator(rs.getInt("operator"))
            .deliveryTime(rs.getTimestamp("delivery_time")==null?null:rs.getTimestamp("delivery_time").toLocalDateTime())
            .attachment(rs.getString("attachment"))
            .clientOrderGoodsId(rs.getString("client_order_goods_id"))
            .relationId(rs.getInt("relation_id"))
            .contractGoodsId(rs.getInt("contract_goods_id"))
            .contractService(rs.getBoolean("contract_service"))
            .remindTime(rs.getTimestamp("remind_time")==null?null:rs.getTimestamp("remind_time").toLocalDateTime())
            .returnedNum(rs.getDouble("returned_num"))
            .returnReceivedNum(rs.getDouble("return_received_num"))
            .costCenterName(rs.getString("cost_center_name"))
            .modifiedBy(rs.getString("modified_by"))
            .createdBy(rs.getString("created_by"))
            .demandGoodsId(rs.getInt("demand_goods_id"))
            .clarifyTips(rs.getString("clarify_tips"))
            .returnPmReceivedNum(rs.getDouble("return_pm_received_num"))
            .actualDeliverTime(rs.getTimestamp("actual_deliver_time")==null?null:rs.getTimestamp("actual_deliver_time").toLocalDateTime())
            .organizationSku(rs.getString("organization_sku"))
            .locationNumber(rs.getString("location_number"))
            .build();


    private ApplicationEventPublisher applicationEventPublisher;
    private BusProperties busProperties;
    private JobBuilderFactory jobBuilderFactory;
    private StepBuilderFactory stepBuilderFactory;
    private DataSource dataSource;
    private ElasticSearchRest rest;
    private RestHighLevelClient client;
    private ObjectMapper mapper;

    public OrderGoodsJob(ApplicationEventPublisher applicationEventPublisher, BusProperties busProperties, JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory, RestHighLevelClient client,ObjectMapper mapper, @Qualifier("gomroDataSource") DataSource dataSource) {

        this.applicationEventPublisher = applicationEventPublisher;
        this.busProperties = busProperties;
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.dataSource = dataSource;

        this.client = client;
        this.mapper = mapper;
        this.rest = new ElasticSearchRest(client, mapper);
    }

    @Bean(JobName)
    public Job job(@Qualifier(StepName) Step step) {

        return jobBuilderFactory.get(JobName)
                .incrementer(new RunIdIncrementer())
                .flow(step)
                .end()
                .listener(jobExecutionListener)
                .build();
    }

    @Bean(StepName)
    public Step step() {

        return stepBuilderFactory.get(StepName)
                .<AcquireOrderGoods, AcquireOrderGoods>chunk(100)
                .reader(reader())
                .processor(processor())
                .writer(writer())
                .build();
    }

    @Bean(ReaderName)
    @StepScope
    public JdbcPagingItemReader<AcquireOrderGoods> reader() {

        // 获取最后更新时间
        AcquireLastBatch acquireLastBatch = rest.documentGet(BiAcquireLastName, AcquireOrderGoods.TableName, AcquireLastBatch.class);
        Date last = null;
        if (acquireLastBatch != null && acquireLastBatch.getLast() != null) last = acquireLastBatch.getLast();

        String where = null;
        Map<String, Object> parameterValues = new HashMap<>();
        if (last != null) {
            where = "where (last is not null and last >= :last) or (time >= :time)";
            parameterValues.put("last", last);
            parameterValues.put("time", last);
        }

        return new JdbcPagingItemReaderBuilder<AcquireOrderGoods>()
                .name(ReaderName)
                .dataSource(dataSource)
                .pageSize(100)
                .selectClause("select *")
                .fromClause("from " + AcquireOrderGoods.TableName)
                .whereClause(where)
                .parameterValues(parameterValues)
                .sortKeys(new LinkedHashMap<String, Order>() {

                    private static final long serialVersionUID = 8503566519535362740L;

                    {
                        put("id", Order.ASCENDING);
                    }
                })
                .rowMapper(rowMapper)
                .build();
    }

    public ElasticSearchClzItemWriter<AcquireOrderGoods> writer() {

        return new ElasticSearchClzItemWriter<>(client,mapper, Constants.BiWarehouseIndexPrefix + AcquireOrderGoods.TableName);
    }

    public ItemProcessor<AcquireOrderGoods, AcquireOrderGoods> processor() {

        return item -> item;
    }

    private JobExecutionListener jobExecutionListener = new JobExecutionListener() {

        @Override
        public void beforeJob(@NonNull JobExecution jobExecution) {

            ElasticSearchRest rest = new ElasticSearchRest(client,mapper);

            // 初始化数据索引
            if (!rest.indexExists(BiWarehouseName)) {

                rest.indexCreate(BiWarehouseName);
                rest.indexMappingPut(BiWarehouseName, "acquisition/" + BiWarehouseName + ".json", true);
                rest.indexFlush(BiWarehouseName, null, null);
            }

            // 初始化最后更新时间索引
            if (!rest.indexExists(BiAcquireLastName)) {

                rest.indexCreate(BiAcquireLastName);
                rest.indexMappingPut(BiAcquireLastName, "acquisition/" + BiAcquireLastName + ".json", true);
                rest.indexFlush(BiAcquireLastName, null, null);
            }
        }

        @Override
        public void afterJob(@NonNull JobExecution jobExecution) {

            // 回写最后更新时间
            if (jobExecution.getStatus() == BatchStatus.COMPLETED) {

                rest.documentCreate(Constants.BiWarehouseIndexPrefix + AcquireLastBatch.TableName, AcquireOrderGoods.TableName, AcquireLastBatch.builder().id(AcquireOrderGoods.TableName).last(jobExecution.getCreateTime()).build());

                applicationEventPublisher.publishEvent(new BiAcquireBatchEvent(this, busProperties.getId(), "**", AcquirePlan.OrderGoodsPlan, AcquireBatchStatus.JobCompleted));
            }
        }
    };
}
